package main

import (
	"fmt"
	"github.com/Shopify/sarama"
	"github.com/Unknwon/goconfig"
	log "socketTokafka/logger"
	kafka "socketTokafka/product"
	"socketTokafka/server"
)

func main() {

	cfg, err := goconfig.LoadConfigFile("conf.ini")
	if err != nil {
		log.Logger.Error("读取配置文件出错")
	}
	broker, err := cfg.GetValue("kafka", "broker")
	if err != nil {
		log.Logger.Error("kafka broker 参数读取失败")
	}
	frequency, err := cfg.Int("kafka", "frequency")
	if err != nil {
		log.Logger.Error("kafka frequency 参数读取失败")
	}
	maxmessage, err := cfg.Int("kafka", "maxmessage")
	if err != nil {
		log.Logger.Error("kafka maxmessage 参数读取失败")
	}
	topicA, err := cfg.GetValue("kafka", "topicA")
	if err != nil {
		log.Logger.Error("kafka topicA 参数读取失败")
	}
	topicB, err := cfg.GetValue("kafka", "topicB")
	if err != nil {
		log.Logger.Error("kafka topicB 参数读取失败")
	}
	goroutineNumber, err := cfg.Int("goroutine", "goroutineNumber")
	if err != nil {
		log.Logger.Error("kafka goroutineNumber 参数读取失败")
	}

	ca := &kafka.Config{
		topicA,
		topicB,
		broker,
		frequency,
		maxmessage,
	}

	product, err := kafka.NewProducer(ca)
	if err != nil {
		log.Logger.Error(err)
	}
	mqa := make(chan *sarama.ProducerMessage, maxmessage)
	product.SetMsgaQ(mqa)
	mqb := make(chan *sarama.ProducerMessage, maxmessage)
	product.SetMsgbQ(mqb)

	address, err := cfg.GetValue("udpserver", "address")
	if err != nil {
		log.Logger.Error("udpserver address 参数读取失败")
	}
	port, err := cfg.GetValue("udpserver", "port")
	if err != nil {
		log.Logger.Error("udpserver port 参数读取失败")
	}
	udpaddr := fmt.Sprintf("%s:%s", address, port) //s3 = "打印字符串"

	srv := server.NewServer(product, udpaddr, goroutineNumber)
	log.Logger.Info("启动 udp 服务监听-----------")
	srv.UdpServer(udpaddr, mqa, mqb)
	srv.GetClient().Run()
	if err != nil {
		log.Logger.Error("启动失败")
	}

	defer srv.GetClient().Close()
}
